//
// Created by 廖海龙 on 2025/4/8.
//

#include "HelloVideoCacheServer.hpp"


HelloVideoCacheServer::HelloVideoCacheServer(const char *tag) : logger(tag), config(),
                                                                cache(nullptr),
                                                                bitstream(nullptr),
                                                                playlist(nullptr),
                                                                port(-1),
                                                                listenFd(-1),
                                                                pool(),
                                                                running(false),
                                                                serverThread(nullptr),
                                                                clientBuffers()
{
    logger.i("HelloVideoCacheServer::HelloVideoCacheServer(%p)", this);
}

HelloVideoCacheServer::~HelloVideoCacheServer()
{
    stop();
    logger.i("HelloVideoCacheServer::~HelloVideoCacheServer(%p)", this);
}

void HelloVideoCacheServer::setConfig(const HelloCacheConfig &c)
{
    if (listenFd > 0 || port > 0)
    {
        logger.i("server is running please call stop() first");
        return;
    }
    this->config = c;
    this->cache = std::make_shared<HelloCacheManager>(c.cacheDir);
    this->bitstream = std::make_shared<HelloBitStreamHandler>(this->cache);
    this->playlist = std::make_shared<HelloPlaylistHandler>(this->cache);
    this->playlist->setUrlTransformer(HelloVideoCacheServer::onUrlTransformer, this);
    logger.i("root cache dir: %s", c.cacheDir.c_str());
}

void HelloVideoCacheServer::start()
{
    if (running)
    {
        logger.i("server is started please call stop() first");
        return;
    }

    running = true;

    // 开启子线程，启动服务监听
    serverThread = new std::thread(&HelloVideoCacheServer::onClientServerReadable,
                                   std::ref(*this));
}

std::string HelloVideoCacheServer::getProxyUrl(const std::string &url)
{
    if (running && port > 0 && pool.clients < FD_SETSIZE - 1) // 控制最大请求1023个,其中有一个是服务端的fd
    {
        const std::string &_url = HelloHttpRequest::urlEncode(url);
        return "http://127.0.0.1:" + std::to_string(port) + "/hellovideocache?url=" + _url;
    }
    return url;
}

void HelloVideoCacheServer::stop()
{

    // 所有子线程都需要阻塞式释放掉
    HelloServerHelper::releaseThread(&serverThread);
    // 释放所有客户端资源逻辑线程
    if (playlist)
    {
        playlist->clearAndWaitAllFetcher();
    }
    if (bitstream)
    {
        bitstream->clearAndWaitAllFetcher();
    }

    // 还原属性字段
    if (listenFd > 0)
    {
        FD_CLR(listenFd, &pool.read_set); // 在关闭前移除
        close(listenFd);
        listenFd = -1;
    }
    port = -1;
    releasePool(&pool);

    running = false;
}


bool HelloVideoCacheServer::prepare()
{
    // 获取一个可用的端口
    int from = this->config.defaultPort;
    int to = 65535;
    int bindPort = -1;
    int fd = -1;
    for (int p = from; p < to; ++p)
    {
        std::string server = std::to_string(p);
        fd = open_listenfd(server.c_str());
        if (fd > 0)
        {
            bindPort = p;
            break;
        }
    }

    if (bindPort <= 0)
    {
        logger.i("prepare failure cannot bind a free port");
        return false;
    }

    this->port = bindPort;
    this->listenFd = fd;

    initPool(&pool, fd);

    return true;
}

/**
 * @param p 初始化连接池
 */
void HelloVideoCacheServer::initPool(Pool *p, int listenFd)
{
    // Initially, there are no connected descriptors
    int i;
    p->maxi = -1;
    p->clients = 0;
    for (i = 0; i < FD_SETSIZE; ++i)
    {
        p->client_fds[i] = -1;
    }

    // Initially, listenfd is only member of select read set
    p->maxfd = listenFd;
    FD_ZERO(&p->read_set);
    FD_SET(listenFd, &p->read_set);
}

void HelloVideoCacheServer::releasePool(Pool *p)
{
    for (int i = 0; i <= p->maxi; ++i)
    {
        removeClient(p, i);
    }
    FD_ZERO(&p->read_set);
    p->maxfd = -1;
    p->maxi = -1;
    p->nb_ready = 0;
}

/**
 * @param p 连接池
 * @param connFd 新加入的
 * @return true表示添加成功
 */
bool HelloVideoCacheServer::addClient(Pool *p, int connFd)
{
    logger.i("add client fd[%d]", connFd);
    int i;
    p->nb_ready--; // Number of ready descriptors from select
    for (i = 0; i < FD_SETSIZE; i++) // Find an available slot
    {
        if (p->client_fds[i] < 0)
        {
            // Add connected descriptor to the pool
            p->client_fds[i] = connFd;
            rio_readinitb(&p->client_rio[i], connFd);

            // Add the descriptor to descriptor set
            FD_SET(connFd, &p->read_set);

            // Update max descriptor and pool high water mark
            if (connFd > p->maxfd)
                p->maxfd = connFd;
            if (i > p->maxi)
                p->maxi = i;

            p->clients++;
            logger.i("success add client on fd %d, and maxi is %d", connFd, p->maxi);

            break;
        }
    }

    if (i == FD_SETSIZE)
    {
        logger.i("add_client error: Too many clients");
        return false;
    }

    // 为每个客户端添加一个buffer，接收http请求头数据
    clientBuffers[connFd] = "";

    return true;
}

void HelloVideoCacheServer::removeClient(Pool *p, int i)
{
    int connFd = p->client_fds[i];
    if (connFd < 0)return;
    logger.i("remove client fd[%d]", connFd);

    clientBuffers.erase(connFd);
    close(connFd);
    FD_CLR(connFd, &p->read_set);
    p->client_fds[i] = -1;
    p->clients--;
    // 客户端连接断开, 移除+停止 fetcher 防止无用功
    if (playlist)
    {
        playlist->removeDataFetcher(connFd);
    }
    if (bitstream)
    {
        bitstream->removeDataFetcher(connFd);
    }
}

/**
 * @param p 检查 pool 中客户端可读事件
 */
void HelloVideoCacheServer::checkClients(Pool *p)
{
    int i, connFd;
    long n;
    char buf[MAXLINE];
    rio_t rio;

    for (i = 0; (i <= p->maxi) && (p->nb_ready > 0); i++)
    {
        connFd = p->client_fds[i];
        rio = p->client_rio[i];

        // If descriptor is ready, echo a text line from it
        if ((connFd > 0) && (FD_ISSET(connFd, &p->ready_set)))
        {
            p->nb_ready--;

            std::string &buffer = clientBuffers[connFd];
            if ((n = rio_read(&rio, buf, MAXLINE)) > 0)
            {
                // 追加到buffer中，一个客户端的http请求内容可能很长
                buffer.append(buf, n);
                logger.i("Received[fd:%d,buffer:%d,n:%d]", connFd, buffer.size(), n);
                // 处理HTTP请求首部字段数据
                if (buffer.find("\r\n\r\n") != std::string::npos)
                {
                    logger.i("Received OK[fd:%d]: \n%s", connFd, buffer.c_str());

                    // 示例：
                    //    GET /proxy?url=http://demo-videos.qnsdk.com/VR-Panorama-Equirect-Angular-4500k.mp4 HTTP/1.1\r\n
                    //    User-Agent: Lavf/61.7.100\r\n
                    //    Accept: */*\r\n
                    //    Range: bytes=0-\r\n
                    //    Connection: close\r\n
                    //    Host: 10.131.40.146:8898\r\n
                    //    Icy-MetaData: 1\r\n
                    //    \r\n

                    // 处理http请求头的内容，整理出 query + headers
                    std::shared_ptr <HelloHttpRequest> request =
                            std::make_shared<HelloHttpRequest>(connFd, buffer);
                    onClientRequest(request);

                    // 重置buffer，减少内存
                    buffer = "";
                }

            } else
            { // EOF detected, remove descriptor from pool
                // do something to remove client
                logger.i("EOF remove client fd %d\n", connFd);
                removeClient(&pool, i);
            }
        }
    }
}


void HelloVideoCacheServer::onClientServerReadable()
{
    if (!prepare())
    {
        logger.e("prepare is not success");
        return;
    }

    logger.i("listening client connection is start port[%d]", port);
    struct sockaddr_storage client_address{};
    socklen_t client_len = sizeof(struct sockaddr_storage);

    // 这个子线程专门阻塞式的来获取
    while (running || listenFd >= 0)
    {
        // Wait for listening/connected descriptor(s) to become ready
        pool.ready_set = pool.read_set;
        int ret = select(pool.maxfd + 1, &pool.ready_set, nullptr, nullptr, nullptr);
        if (ret < 0)
        {
            if (errno == EBADF || errno == EINVAL || listenFd < 0)
            {
                logger.i("server's listen fd is closed");
                break;
            }
            logger.i("select return [%d]", ret);
        }
        pool.nb_ready = ret; // 记录当前可读句柄个数

        // If listening descriptor ready, add new client to pool
        if (FD_ISSET(listenFd, &pool.ready_set))
        {
            int connFd = accept(listenFd, (SA *) &client_address, &client_len);
            addClient(&pool, connFd); // 服务端接收到客户端的请求连接，保存起来，等待下一次select监听可读事件
        }
        // deal with client's input
        checkClients(&pool); // 这次来的可读事件里面，可能还有客户端的，一并检查
    }

    // 服务器是因为主动关闭也好，还是异常关闭也好，总之都需要把所有客户端相关资源都释放掉
    releasePool(&pool);
    logger.i("listening client connection is over");
}

void HelloVideoCacheServer::onClientRequest(const std::shared_ptr <HelloHttpRequest> &request)
{
    const int fd = request->getConnFd();
    const std::string &url = request->getRealUrl();
    if (url.empty())
    {
        logger.e("client[%d] has wrong url", fd);
        return;
    }

    std::string path = url.substr(0, url.find('?'));

    if (HelloHeader::endsWith(path, ".m3u8"))
    {
        // 处理 播放列表 式的请求
        if (playlist)
        {
            logger.i("client[%d] playlist->acceptClient[%s]", fd, url.c_str());
            playlist->acceptClient(request);
        }
    } else if (HelloHeader::endsWith(path, ".mp4") || HelloHeader::endsWith(path, ".ts"))
    {
        // 处理 比特流 式的请求
        if (bitstream)
        {
            logger.i("client[%d] bitstream->acceptClient[%s]", fd, url.c_str());
            bitstream->acceptClient(request);
        }
    } else
    {
        // url类型不匹配
        close(fd);
        logger.e("client[%d] cannot handle url[%s]", fd, url.c_str());
    }

}

std::string HelloVideoCacheServer::onUrlTransformer(const std::string &uri, void *userdata)
{
    auto native = reinterpret_cast<HelloVideoCacheServer *>(userdata);
    return native->getProxyUrl(uri);
}


